package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/13
 */
public class App05_DwdMhsTestReport extends BaseSqlApp {
    public static void main(String[] args) {
        new App05_DwdMhsTestReport().init(
                5005,
                "App5_DwdMhsTestReport",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "REQUISITION_ID                    STRING," +
                "HOSPITAL_NAME                     STRING," +
                "HOSPITAL_CODE                     STRING," +
                "AGE_UNIT                          STRING," +
                "AGE                               STRING," +
                "SEX_NAME                          STRING," +
                "SEX_CODE                          STRING," +
                "ID_CARD_NUMBER                    STRING," +
                "PATIENT_VISIT_CATEGORY_CODE       STRING," +
                "PATIENT_VISIT_CATEGORY            STRING," +
                "INPATIENT_VISIT_FLOW_ID           STRING," +
                "INPATIENT_NO                      STRING," +
                "OUTPATIENT_VISIT_FLOW_ID          STRING," +
                "OUTPATIENT_NO                     STRING," +
                "PATIENT_ID                        STRING," +
                "PATIENT_NAME                      STRING," +
                "DOMAIN_ID                         STRING," +
                "VISIT_TIMES                       STRING," +
                "VISIT_DATE                        STRING," +
                "TEST_REPORT_NO                    STRING," +
                "APPLY_SHEET_NO                    STRING," +
                "SPECIMEN_NUMBER                   STRING," +
                "TELECOM                           STRING," +
                "TEST_DATE                         STRING," +
                "TEST_REPORT_DOCTOR_ID             STRING," +
                "TEST_REPORT_DOCTOR_NAME           STRING," +
                "AUTHOR_CODE                       STRING," +
                "AUTHOR_NAME                       STRING," +
                "CUSTODIAN_NAME                    STRING," +
                "CUSTODIAN_CODE                    STRING," +
                "AUDIT_SIGNDATE                    STRING," +
                "TEST_AUDIT_DOCTOR_ID              STRING," +
                "TEST_AUDIT_DOCTOR_NAME            STRING," +
                "CHECK_SIGNDATE                    STRING," +
                "TEST_TECH_STAFF_ID                STRING," +
                "TEST_TECH_STAFF_NAME              STRING," +
                "TECHNICIAN_SIGNDATE               STRING," +
                "TEST_DOCTOR_ID                    STRING," +
                "TEST_DOCTOR_NAME                  STRING," +
                "APPLY_DATE                        STRING," +
                "BED_NO                            STRING," +
                "BED_NAME                          STRING," +
                "ROOM_NO                           STRING," +
                "ROOM_NAME                         STRING," +
                "DEPT_ID                           STRING," +
                "DEPT_NAME                         STRING," +
                "WARDS_ID                          STRING," +
                "WARDS_NAME                        STRING," +
                "INSPECTION_METHOD                 STRING," +
                "INSPECT_TYPE                      STRING," +
                "TEST_REPORT_RESULT                STRING," +
                "REPORT_DEPT                       STRING," +
                "AGENCY_NAME                       STRING," +
                "TEST_REPORT_COOMMENTS             STRING," +
                "INSPECT_TYPE_CODE                 STRING," +
                "INSPECTION_ITEM                   STRING," +
                "APPLY_MED_ORG_ID                  STRING," +
                "TEST_REPORT_DEPT_ID               STRING," +
                "APPLY_DEPT_ID                     STRING," +
                "APPLY_DEPT_NAME                   STRING," +
                "SPECIMEN_CLASS                    STRING," +
                "SPECIMEN_STATUS                   STRING," +
                "DIAG_CODE                         STRING," +
                "DIAG_NAME                         STRING," +
                "DIAG_DATE                         STRING," +
                "VISIT_ID                          STRING," +
                "primary key (REQUISITION_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_TEST_REPORT));
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id as REQUISITION_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "case when patient_typecode = '3' then visit_id else '' end as INPATIENT_VISIT_FLOW_ID , " +
                "inpatient as INPATIENT_NO , " +
                "case when patient_typecode <> '3' then visit_id else '' end as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "visit_times as VISIT_TIMES , " +
                "visit_date as VISIT_DATE , " +
                "report_id as TEST_REPORT_NO , " +
                "appy_id as APPLY_SHEET_NO , " +
                "specimen_number as SPECIMEN_NUMBER , " +
                "telecom as TELECOM , " +
                "write_time as TEST_DATE , " +
                "author_code as TEST_REPORT_DOCTOR_ID , " +
                "author_name as TEST_REPORT_DOCTOR_NAME , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "audit_signdate as AUDIT_SIGNDATE , " +
                "audit_code as TEST_AUDIT_DOCTOR_ID , " +
                "audit_name as TEST_AUDIT_DOCTOR_NAME , " +
                "check_signdate as CHECK_SIGNDATE , " +
                "technician_code as TEST_TECH_STAFF_ID , " +
                "technician_name as TEST_TECH_STAFF_NAME , " +
                "technician_signdate as TECHNICIAN_SIGNDATE , " +
                "examiner_code as TEST_DOCTOR_ID , " +
                "examiner_name as TEST_DOCTOR_NAME , " +
                "apply_date as APPLY_DATE , " +
                "bed_no as BED_NO , " +
                "bed_name as BED_NAME , " +
                "ward_id as ROOM_NO , " +
                "ward_name as ROOM_NAME , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "wards_id as WARDS_ID , " +
                "wards_name as WARDS_NAME , " +
                "inspection_method as INSPECTION_METHOD , " +
                "inspect_type as INSPECT_TYPE , " +
                "report_result as TEST_REPORT_RESULT , " +
                "report_dept as REPORT_DEPT , " +
                "agency_name as AGENCY_NAME , " +
                "report_remark as TEST_REPORT_COOMMENTS , " +
                "inspect_type_code as INSPECT_TYPE_CODE , " +
                "inspection_item as INSPECTION_ITEM , " +
                "apply_agencycode as APPLY_MED_ORG_ID , " +
                "apply_organ as TEST_REPORT_DEPT_ID , " +
                "apply_dept_code as APPLY_DEPT_ID , " +
                "apply_dept as APPLY_DEPT_NAME , " +
                "specimen_category as SPECIMEN_CLASS , " +
                "specimen_status as SPECIMEN_STATUS , " +
                "diag_code as DIAG_CODE , " +
                "diag_name as DIAG_NAME , " +
                "diag_date as DIAG_DATE ," +
                "visit_id as VISIT_ID " +
                "from hdsd00_05_01 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    " +
                "left join tdy_list_exam e on s.unique_id = e.unique_id and s.xds_type = e.xds_type    ");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_05_01
        tableEnv.executeSql("create table hdsd00_05_01 (" +
                "pk                       string," +
                "upload_time              string," +
                "status                   string," +
                "empi                     string," +
                "encounter_id             string," +
                "bed_name                 string," +
                "ward_name                string," +
                "wards_id                 string," +
                "patient_id               string," +
                "patient_domain           string," +
                "visit_domain             string," +
                "visit_id                 string," +
                "visit_times              string," +
                "unique_id                string," +
                "xds_type                 string," +
                "xds_name                 string," +
                "effective_time           string," +
                "xds_version              string," +
                "out_patient_id           string," +
                "domain_id                string," +
                "visit_date               string," +
                "inpatient                string," +
                "patient_type             string," +
                "report_id                string," +
                "patient_typecode         string," +
                "appy_id                  string," +
                "specimen_number          string," +
                "telecom                  string," +
                "identify_no              string," +
                "patient_name             string," +
                "sex_code                 string," +
                "sex                      string," +
                "age                      string," +
                "age_unit                 string," +
                "write_time               string," +
                "author_code              string," +
                "author_name              string," +
                "custodian_name           string," +
                "custodian_code           string," +
                "audit_signdate           string," +
                "audit_code               string," +
                "audit_name               string," +
                "check_signdate           string," +
                "technician_code          string," +
                "technician_name          string," +
                "technician_signdate      string," +
                "examiner_code            string," +
                "examiner_name            string," +
                "apply_date               string," +
                "apply_dept_code          string," +
                "apply_dept               string," +
                "apply_agencycode         string," +
                "apply_organ              string," +
                "bed_no                   string," +
                "ward_id                  string," +
                "dept_code                string," +
                "dept_name                string," +
                "wards_name               string," +
                "hospital_code            string," +
                "hospital_name            string," +
                "inspection_method        string," +
                "inspect_type             string," +
                "report_result            string," +
                "report_dept              string," +
                "agency_name              string," +
                "report_remark            string," +
                "inspect_type_code        string," +
                "inspection_item          string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_05_01","App5_DwdMhsTestReport"));
        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App5_DwdMhsTestReport");
        // tdy_list_exam
        super.readTdyListExam(tableEnv,"App5_DwdMhsTestReport");



    }
}
